package edu.mbb.cloud.nacos.listen;

import edu.mbb.cloud.nacos.NacosProperSource;
import edu.mbb.cloud.nacos.SimpleHttpUtil;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 监听的主方法，所有的更新和监听都在此方法完成
 */
public class ClientWorker {



    //我们不考虑移除
//    public void removeListener(String dataId, String group, Listener listener) {
//        CacheData cache = getCache(dataId, group);
//        if (null != cache) {
//            cache.removeListener(listener);
//            if (cache.getListeners().isEmpty()) {
//                removeCache(dataId, group);
//            }
//        }
//    }



    /**
     * 添加监听器。。也就是存储对应的监听对应的 配置文件  源代码是使用 这个类型的包裹了map类型AtomicReference
     * @param dataId
     * @param group
     * @param content
     * @param listeners
     */
    public void addTenantListenersWithContent(String dataId, String group, String content, List<? extends Listener> listeners) throws Exception {
        //nameSpace
        String nameSpace = properties.getProperty("namespace");
        CacheData cache = addCacheDataIfAbsent(dataId, group, nameSpace);
        cache.setContent(content);
        for (Listener listener : listeners) {
            cache.addListener(listener);
        }
    }


    public CacheData addCacheDataIfAbsent(String dataId, String group, String tenant) throws Exception {
        CacheData cache = getCache(dataId, group, tenant);
        if (null != cache) {
            return cache;
        }
        String key = getKeyTenant(dataId, group, tenant);
        synchronized (cacheMap) {
            CacheData cacheFromMap = getCache(dataId, group, tenant);
            // multiple listeners on the same dataid+group and race condition,so
            // double check again
            // other listener thread beat me to set to cacheMap
            if (null != cacheFromMap) {
                cache = cacheFromMap;
            } else {
                // 本地没有文件
                String content ="";
                if (enableRemoteSyncConfig) {
                    //从nacos 上面拿配置
                     content = getServerConfig(tenant,dataId, group );

                }
                cache = new CacheData(tenant, dataId, group,content);
            }

            Map<String, CacheData> copy = new HashMap<String, CacheData>(cacheMap.get());
            copy.put(key, cache);
            cacheMap.set(copy);
        }
        System.out.println("[{}] [subscribe] {}"+key);

        //暂不实现
        //MetricsMonitor.getListenConfigCountMonitor().set(cacheMap.get().size());

        return cache;
    }

    public String getServerConfig(String tenant, String dataId, String group){
        String serverAddr = properties.getProperty("serverAddr");
        String url1="http://"+serverAddr+"/nacos/v1/cs/configs?dataId="+dataId+"&group="+group+"&tenant="+tenant+"&namespaceId="+tenant;
        String s = SimpleHttpUtil.methodGet(url1);
        return s;
    }


    public CacheData getCache(String dataId, String group, String tenant) {
        if (null == dataId || null == group) {
            throw new IllegalArgumentException();
        }
        return cacheMap.get().get(getKeyTenant(dataId, group, tenant));
    }


    /**
     * 检查本地的配置文件 我们不模拟本地的情况这个东西 本地情况不考虑
     * 主要是提供了 没有-》有 。。。。一些处理的过程，说明本地的配置文件可以动态添加的。
     */
   /* private void checkLocalConfig(CacheData cacheData) {
        final String dataId = cacheData.dataId;
        final String group = cacheData.group;
        final String tenant = cacheData.tenant;
        File path = LocalConfigInfoProcessor.getFailoverFile(agent.getName(), dataId, group, tenant);

        // 没有 -> 有
        if (!cacheData.isUseLocalConfigInfo() && path.exists()) {
            String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);
            String md5 = MD5.getInstance().getMD5String(content);
            cacheData.setUseLocalConfigInfo(true);
            cacheData.setLocalConfigInfoVersion(path.lastModified());
            cacheData.setContent(content);

            LOGGER.warn("[{}] [failover-change] failover file created. dataId={}, group={}, tenant={}, md5={}, content={}",
                    agent.getName(), dataId, group, tenant, md5, ContentUtils.truncateContent(content));
            return;
        }

        // 有 -> 没有。不通知业务监听器，从server拿到配置后通知。
        if (cacheData.isUseLocalConfigInfo() && !path.exists()) {
            cacheData.setUseLocalConfigInfo(false);
            LOGGER.warn("[{}] [failover-change] failover file deleted. dataId={}, group={}, tenant={}", agent.getName(),
                    dataId, group, tenant);
            return;
        }

        // 有变更
        if (cacheData.isUseLocalConfigInfo() && path.exists()
                && cacheData.getLocalConfigInfoVersion() != path.lastModified()) {
            String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);
            String md5 = MD5.getInstance().getMD5String(content);
            cacheData.setUseLocalConfigInfo(true);
            cacheData.setLocalConfigInfoVersion(path.lastModified());
            cacheData.setContent(content);
            LOGGER.warn("[{}] [failover-change] failover file changed. dataId={}, group={}, tenant={}, md5={}, content={}",
                    agent.getName(), dataId, group, tenant, md5, ContentUtils.truncateContent(content));
        }
    }*/

   //得到对应的group 我们不允许为空就行了。。所以也不要这个方法了
/*    private String null2defaultGroup(String group) {
        return (null == group) ? Constants.DEFAULT_GROUP : group.trim();
    }*/

    //开始分任务的检查对应的配置文件是否变动了。
    public void checkConfigInfo() {
        System.out.println("当前启动第一个线程"+Thread.currentThread().getName());
        // 分任务 nacos本身都没实现，我们也就不实现了 默认3000个分一批
        //一般我们也没有那个多。。。。我们就不用分任务
        //进行对应的直接取配置文件的执行
        int listenerSize = cacheMap.get().size();
        // 向上取整为批数
           // for (int i = 0; i < listenerSize; i++) {
                // 要判断任务是否在执行 这块需要好好想想。 任务列表现在是无序的。变化过程可能有问题
                executorService.execute(new LongPollingRunnable(0));
           // }
    }

    public static final String LINE_SEPARATOR = Character.toString((char) 1);

    public static final String WORD_SEPARATOR = Character.toString((char) 2);
    /**
     * 从Server获取值变化了的DataID列表。返回的对象里只有dataId和group是有效的。 保证不返回NULL。
     */
    //拼装所有的请求的配置文件 我们就不拼装那麻烦了 直接遍历进行拼装就行了
    List<String> checkUpdateDataIds(List<CacheData> cacheDatas) throws IOException {
        StringBuilder sb = new StringBuilder();
        for (CacheData cacheData : cacheDatas) {
            //我们不使用本地 取消
          //  if (!cacheData.isUseLocalConfigInfo()) {
            sb.append(cacheData.getDataId()).append(WORD_SEPARATOR);
            sb.append(cacheData.getGroup()).append(WORD_SEPARATOR);
            sb.append(cacheData.getMd5()).append(WORD_SEPARATOR);
            sb.append(cacheData.getTenant()).append(LINE_SEPARATOR);
        }
        return checkUpdateConfigStr(sb.toString());
    }
    public static final String PROBE_MODIFY_REQUEST = "Listening-Configs";

    public static final String CONFIG_CONTROLLER_PATH = "/nacos/v1/cs/configs";
    /**
     * 从Server获取值变化了的DataID列表。返回的对象里只有dataId和group是有效的。 保证不返回NULL。
     */
    List<String> checkUpdateConfigStr(String probeUpdateString) throws IOException {

        //List<String> params = Arrays.asList(PROBE_MODIFY_REQUEST, probeUpdateString);

        if (StringUtils.isEmpty(probeUpdateString)) {
            return Collections.emptyList();
        }
        try {
            //参数长成这样。
            // Listening-Configs=nacos-config-client-dev.yaml%02TEST_GROUP%024706e98b2d9447ddc23dee45b3d01d01%02b8b51e18-7395-4124-8840-fe39fde231b1%01nacos-config-client.yaml%02TEST_GROUP%02a9d26e4b7c8ab5043dd9c0fb2214bf1c%02b8b51e18-7395-4124-8840-fe39fde231b1%01
            //nacos有重试机制 重试 3次。我就不写这块的逻辑 就是 失败了尝试3次 3次还是失败 就算真正 的失败
            String addr = properties.getProperty("serverAddr");
            String url="http://"+addr+CONFIG_CONTROLLER_PATH+"/listener";

            Map<String, String> headMap=new HashMap<>();
            headMap.put("Long-Pulling-Timeout","4500");
            headMap.put("Content-Type", "application/x-www-form-urlencoded;charset=UTF-8");
            probeUpdateString= URLEncoder.encode(probeUpdateString);
            String result = SimpleHttpUtil.methodPost(url, headMap, PROBE_MODIFY_REQUEST + "=" + probeUpdateString);

            if(StringUtils.isEmpty(result)) {
                //说明没有更新
                //设置健康状态
                setHealthServer(true);
                return Collections.emptyList();
            }else{
                //假如不为空说明有更新 进行对象的消息推送
                setHealthServer(true);
                return parseUpdateDataIdResponse(result);
            }

        } catch (Exception e) {
            setHealthServer(false);
            e.printStackTrace();
            throw e;
        }
    }
    static public String getKey(String dataId, String group) {
        StringBuilder sb = new StringBuilder();
        urlEncode(dataId, sb);
        sb.append('+');
        urlEncode(group, sb);
        return sb.toString();
    }
    /**
     * + -> %2B % -> %25
     */
    static void urlEncode(String str, StringBuilder sb) {
        for (int idx = 0; idx < str.length(); ++idx) {
            char c = str.charAt(idx);
            if ('+' == c) {
                sb.append("%2B");
            } else if ('%' == c) {
                sb.append("%25");
            } else {
                sb.append(c);
            }
        }
    }
    static public String getKeyTenant(String dataId, String group, String tenant) {
        StringBuilder sb = new StringBuilder();
        urlEncode(dataId, sb);
        sb.append('+');
        urlEncode(group, sb);
        if (!StringUtils.isEmpty(tenant)) {
            sb.append('+');
            urlEncode(tenant, sb);
        }
        return sb.toString();
    }
    /**
     * 从HTTP响应拿到变化的groupKey。保证不返回NULL。
     */
    private List<String> parseUpdateDataIdResponse(String response) {
        if (StringUtils.isEmpty(response)) {
            return Collections.emptyList();
        }

        try {
            response = URLDecoder.decode(response, "UTF-8");
        } catch (Exception e) {
            e.printStackTrace();
        }

        List<String> updateList = new LinkedList<String>();

        for (String dataIdAndGroup : response.split(LINE_SEPARATOR)) {
            if (!StringUtils.isEmpty(dataIdAndGroup)) {
                String[] keyArr = dataIdAndGroup.split(WORD_SEPARATOR);
                String dataId = keyArr[0];
                String group = keyArr[1];
                if (keyArr.length == 2) {
                    updateList.add(getKey(dataId, group));
                    System.out.println("[{}] [polling-resp] config changed. dataId="+dataId+", group="+group);
                } else if (keyArr.length == 3) {
                    String tenant = keyArr[2];
                    updateList.add(getKeyTenant(dataId, group, tenant));
                    System.out.println("[{}] [polling-resp] config changed. dataId="+dataId+", group="+group+", tenant="+ tenant);
                } else {
                    System.err.println(dataIdAndGroup+"[{}] [polling-resp] invalid dataIdAndGroup error {}"+dataIdAndGroup);
                }
            }
        }
        return updateList;
    }

    private Properties properties;
    @SuppressWarnings("PMD.ThreadPoolCreationRule")
    public ClientWorker(Properties properties) {
        this.properties=properties;
        //初始化超时参数 我们不管他
        //init(properties);
        //关键部分 启动线程池进行
        executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("com.alibaba.nacos.client.Worker.");
                t.setDaemon(true);
                return t;
            }
        });

        executorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("com.alibaba.nacos.client.Worker.longPolling.");
                t.setDaemon(true);
                return t;
            }
        });
        //任务执行的间隔的设置
        //给对应的任务线程池放任务了
        executor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    checkConfigInfo();
                } catch (Throwable e) {
                    e.printStackTrace();
                    System.out.println("[sub-check] rotate check error");
                }
            }
        }, 10L, 20L, TimeUnit.SECONDS);
    }

    static public String[] parseKey(String groupKey) {
        StringBuilder sb = new StringBuilder();
        String dataId = null;
        String group = null;
        String tenant = null;

        for (int i = 0; i < groupKey.length(); ++i) {
            char c = groupKey.charAt(i);
            if ('+' == c) {
                if (null == dataId) {
                    dataId = sb.toString();
                    sb.setLength(0);
                } else if (null == group) {
                    group = sb.toString();
                    sb.setLength(0);
                } else {
                    throw new IllegalArgumentException("invalid groupkey:" + groupKey);
                }
            } else if ('%' == c) {
                char next = groupKey.charAt(++i);
                char nextnext = groupKey.charAt(++i);
                if ('2' == next && 'B' == nextnext) {
                    sb.append('+');
                } else if ('2' == next && '5' == nextnext) {
                    sb.append('%');
                } else {
                    throw new IllegalArgumentException("invalid groupkey:" + groupKey);
                }
            } else {
                sb.append(c);
            }
        }

        if (StringUtils.isEmpty(group)) {
            group = sb.toString();
            if (group.length() == 0) {
                throw new IllegalArgumentException("invalid groupkey:" + groupKey);
            }
        } else {
            tenant = sb.toString();
            if (group.length() == 0) {
                throw new IllegalArgumentException("invalid groupkey:" + groupKey);
            }
        }

        return new String[] {dataId, group, tenant};
    }
    class LongPollingRunnable implements Runnable {
        private int taskId;

        public LongPollingRunnable(int taskId) {
            this.taskId = taskId;
        }

        @Override
        public void run() {

            List<CacheData> cacheDatas = new ArrayList<CacheData>();
            try {
                // check failover config
                for (CacheData cacheData : cacheMap.get().values()) {
                        cacheDatas.add(cacheData);
                        //我们不管本地的
                }
                // check server config
                List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas);

                for (String groupKey : changedGroupKeys) {
                    String[] key = parseKey(groupKey);
                    String dataId = key[0];
                    String group = key[1];
                    String tenant = null;
                    if (key.length == 3) {
                        tenant = key[2];
                    }
                    try {
                        String content = getServerConfig(tenant,dataId, group);
                        CacheData cache = cacheMap.get().get(getKeyTenant(dataId, group, tenant));
                        cache.setContent(content);
                        System.out.println("监听到变化，重新set 对应的内容");
                    } catch (Exception ioe) {
                        String message = String.format(
                                "[get-update] get changed config exception. dataId=%s, group=%s, tenant=%s", dataId, group, tenant);
                        System.out.println(message);
                    }
                }
                //重新校验md5 假如上一步set内容有变 就会触发事件的推送
                for (CacheData cacheData : cacheDatas) {
                    boolean b = cacheData.checkListenerMd5();
                    if(b){
                        executor.shutdown();
                        executorService.shutdown();
                    }
                    System.out.println("threadName"+Thread.currentThread().getName()+"count:"+b);
                }

                //executorService.execute(this);
            } catch (Throwable e) {

                // If the rotation training task is abnormal, the next execution time of the task will be punished
                System.out.println("出错了");
                e.printStackTrace();
            }
        }
    }

    public boolean isHealthServer() {
        return isHealthServer;
    }

    private void setHealthServer(boolean isHealthServer) {
        this.isHealthServer = isHealthServer;
    }

    final ScheduledExecutorService executor;
    final ScheduledExecutorService executorService;

    /**
     * groupKey -> cacheData
     */
    private final AtomicReference<Map<String, CacheData>> cacheMap = new AtomicReference<Map<String, CacheData>>(
            new HashMap<String, CacheData>());
    private boolean isHealthServer = true;
    private boolean enableRemoteSyncConfig = true;
}
